package com.atguigu.pj.acc

import org.apache.spark.util.AccumulatorV2

import scala.collection.mutable


/**
 * description ：用户点击形为累加器
 * author      ：剧情再美终是戏 
 * mail        : 13286520398@163.com
 * date        ：Created in 2020/1/10 8:22
 * modified By ：
 * version:    : 1.0
 */
class AccUserAction extends AccumulatorV2[(String, String), mutable.Map[(String, String), Long]] {

  // 初始化一个map
  private var map: mutable.Map[(String, String), Long] = mutable.Map[(String, String), Long]()

  // 判断是否零值
  override def isZero: Boolean = map.isEmpty

  // 将数据拷贝到excutor
  override def copy(): AccumulatorV2[(String, String), mutable.Map[(String, String), Long]] = {
    val accUserAction = new AccUserAction()
    accUserAction.map = this.map
    accUserAction
  }

  // 重置map
  override def reset(): Unit = map.empty

  // 累加器数据增加
  override def add(v: (String, String)): Unit = {
    map(v) = map.getOrElse(v, 0L) + 1
  }

  // 从excutor拉取数据回来driver合并
  override def merge(other: AccumulatorV2[(String, String), mutable.Map[(String, String), Long]]): Unit = other match {
    case accUserAction: AccUserAction =>
      val mergeMap = accUserAction.map
      mergeMap.foreach({
        case (k, v) =>
          map(k) = map.getOrElse(k, 0L) + v
      })
  }

  // 返回map的值
  override def value: mutable.Map[(String, String), Long] = map
}
